/*
 *  Copyright 2013 National Institute of Advanced Industrial Science and Technology
 *  
 *  Licensed under the Apache License, Version 2.0 (the "License");
 *  you may not use this file except in compliance with the License.
 *  You may obtain a copy of the License at
 *  
 *      http://www.apache.org/licenses/LICENSE-2.0
 *  
 *  Unless required by applicable law or agreed to in writing, software
 *  distributed under the License is distributed on an "AS IS" BASIS,
 *  WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.
 *  See the License for the specific language governing permissions and
 *  limitations under the License.
 */

package org.sss.common.io;

import java.io.IOException;
import java.util.List;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.sss.common.SentinelThreadPool;
import org.sss.mapreduce.DB;
import org.sss.mapreduce.IOConfiguration;
import org.sss.mapreduce.SssException;
import org.sss.mapreduce.StorageNode;
import org.sss.util.BlockPolicy;

public class IOThread extends Thread {
  private static Logger logger = LoggerFactory.getLogger(IOThread.class);
  private final IOConfiguration ioConf;
  private final int id;
  private final List<StorageNode> storageNodeList;
  private long counter = 0;
  private DBSet dbs = null;
  private StreamDBSet sdbs = null;

  public IOThread(Runnable r,
                  int id,
                  IOConfiguration ioConf,
                  List<StorageNode> storageNodeList,
                  DBSet dbs,
                  boolean isDeamon,
                  String threadName) {
    super(r, String.format("%s-%02d", threadName, id));
    this.ioConf = ioConf;
    this.id = id;
    this.storageNodeList = storageNodeList;

    if (ioConf.stream_protocol) {
      this.sdbs = new StreamDBSet();
    }
    else {
      this.dbs = (dbs == null) ? new DBSet(ioConf.db_factory, ioConf) : dbs;
    }
    if (isDeamon) {
      setDaemon(true);
    }
  }
  public static IOThread currentThread() {
    return (IOThread) Thread.currentThread();
  }
  public int getProcessID() {
    return ioConf.processID;
  }
  public int getThreadID() {
    return id;
  }
  public boolean isRestrictedDBAccess() {
    return ioConf.db_use_vanilla;
  }
  /**
   * .
   * NOTE: thread-unsafe.
   */
  public long getNextCount() {
    return counter++;
  }
  public int getIOGetBatchCount() {
    return ioConf.get_batchcount;
  }
  public int getIOGetLimitNBytes() {
    return ioConf.get_limit_nbytes;
  }
  public int getIOPutBatchCount() {
    return ioConf.put_batchcount;
  }
  public DB getDB(StorageNode node) throws IOException {
    assert(!ioConf.stream_protocol);
    return dbs.get(node);
  }
  public DB getDB(int partition) throws IOException {
    return getDB(storageNodeList.get(partition));
  }
  public StreamDB getStreamDB(StorageNode node) throws SssException {
    assert(ioConf.stream_protocol);
    return sdbs.get(node);
  }
  public StreamDB getStreamDB(int partition) throws SssException {
    return getStreamDB(storageNodeList.get(partition));
  }
  public boolean isStreamDB() throws SssException {
    return ioConf.stream_protocol;
  }

  /* Pool */
  public static SentinelThreadPool createPool(IOConfiguration ioConf,
                                              int threads_max,
                                              int queue_limit,
                                              List<StorageNode> storageNodeList,
                                              String threadName,
                                              boolean shareConnection) {
    return createPool(ioConf, threads_max, queue_limit, storageNodeList, threadName, shareConnection, false);
  }

  public static SentinelThreadPool createPool(IOConfiguration ioConf,
                                              int threads_max,
                                              int queue_limit,
                                              List<StorageNode> storageNodeList,
                                              String threadName,
                                              boolean shareConnection,
                                              boolean isDeamon) {
    return new SentinelThreadPool(
        threads_max, threads_max,
        0L, TimeUnit.MILLISECONDS,
        new LinkedBlockingQueue<Runnable>(queue_limit),
        new Factory(ioConf, storageNodeList, shareConnection, isDeamon, threadName),
        new BlockPolicy());
  }

  private static class Factory implements ThreadFactory {
    private final IOConfiguration ioConf;
    private final List<StorageNode> storageNodeList;
    private final DBSet dbSet;
    private final boolean isDeamon;
    private final String threadName;
    private final AtomicInteger counter = new AtomicInteger();

    public Factory(IOConfiguration ioConf,
                   List<StorageNode> storageNodeList,
                   boolean dbIsShared,
                   boolean isDeamon,
                   String threadName) {
      this.ioConf = ioConf;
      this.storageNodeList = storageNodeList;
      this.isDeamon = isDeamon;

      logger.debug("use shared connection object: .", dbIsShared);
      if (ioConf.stream_protocol && dbIsShared) {
        logger.debug("Can't share StreamDB.");
        dbIsShared = false;
      }
      if (dbIsShared) {
        this.dbSet = new DBSet(ioConf.db_factory, ioConf);
      } else {
        dbSet = null;
      }

      this.threadName = threadName;
    }

    @Override
    public Thread newThread(Runnable r) {
      return new IOThread(r, newId(), ioConf, storageNodeList, dbSet, isDeamon, threadName);
    }

    private int newId() {
      return counter.getAndIncrement();
    }
  }
}
